package com.psbc.api;

import com.google.common.collect.Lists;
import com.psbc.entity.BatchPullResult;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.util.LinkedList;
import java.util.concurrent.TimeoutException;

/**
 * @author:y022 06/08/2022 21:11
 * @description:
 */
@RestController
@Slf4j
@RequestMapping("/rabbitMq")
public class MyRabbitMqTestApi {

    @Autowired
    private ConnectionFactory connectionFactory;

    @GetMapping("/produceMsg")
    public void produceMsg(@RequestParam("msgCount") Integer msgCount) throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("user");
        connectionFactory.setPassword("user");
        connectionFactory.setVirtualHost("/test");
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare("hello-my", true, false, false, null);
        for (int i = 0; i < msgCount; i++) {
            channel.basicPublish("", "hello-my", null, ("hello rabbitmq"+i).getBytes());
        }
        channel.close();
        connection.close();
    }

    @GetMapping("/batchPullMsg")
    public void batchPullMsg() throws IOException, TimeoutException {
        log.info("批量拉取消息启动");
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        MyDeliverCallBack deliverCallback = new MyDeliverCallBack();
        CancelCallback cancelCallback = s -> {
            log.error("发生未知异常!");
        };

        channel.basicConsume("hello-my", true, deliverCallback,cancelCallback);
        BatchPullResult result = deliverCallback.getResult();
        try{
            log.info("成功获取到消息:{}",result.getRes());
//            LockSupport.park();
            channel.basicAck(result.getOffset(),true);
        }catch (Exception e){
            channel.basicNack(result.getOffset(),true,true);
        }
    }

    static class MyDeliverCallBack implements DeliverCallback{
        LinkedList<String> list = Lists.newLinkedList();
        long offset = 0L;
        String  tag ;
        @Override
        public void handle(String s, Delivery delivery) {
            if(list.size()<100){
                list.add(new String(delivery.getBody()));
                offset = delivery.getEnvelope().getDeliveryTag();
                tag = s;
                log.info("msg:",new String(delivery.getBody()));
            }
        }
        public BatchPullResult  getResult(){
            return new BatchPullResult(this.list,this.offset,this.tag);
        }
    }

    }



